package com.urfresh.sp.flume.avro;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.ImmutableMap;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Component
public class MyRpcClientFacade {
    Logger logger = LoggerFactory.getLogger(MyRpcClientFacade.class);

//	private ThreadLocal<RpcClient> client;

    @Value("${flume.hostname}")
    String flumeHostname;
    @Value("${flume.port}")
    int flumePort;

    RpcClient client;

    @PostConstruct
    public void init(){
        client = RpcClientFactory.getDefaultInstance(flumeHostname, flumePort);
    }


    /**
     * 发送到flume中
     *
     * @param data       json数据
     * @param topic      kafka的队里名称
     * @param attachData 附加到json上的数据
     */
    public void sendDataToFlume(String data, String topic, Map<String, String> attachData) {
        // Create a Flume Event object that encapsulates the data
        logger.info("data=" + data + "\r\ntopic=" + topic);

        //如果附加数据不为空，那么就把他解开放入到data中
        data = parseAttachData(attachData, data);

        Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));

        event.setHeaders(ImmutableMap.of("topic", topic));

        // Send the event
        append(event);
    }

    public void sendBatchDataToFlume(List<String> list, String topic, Map<String, String> attachData) {
        List<Event> eventList = new ArrayList<Event>(list.size());

        logger.info("topic=" + topic);
        // Create a Flume Event object that encapsulates the data
        for (String data : list) {
//			//如果附加数据不为空，那么就把他解开放入到data中
            data = parseAttachData(attachData, data);

            logger.info("data=" + data);

            Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
            event.setHeaders(ImmutableMap.of("topic", topic));
            eventList.add(event);
        }
        // Send the batch event
        try {
            client.appendBatch(eventList);
        } catch (EventDeliveryException e) {
            e.printStackTrace();
            logger.error("flume client error, ready to get new instance!");
            synchronized (this) {
                client.close();
                client = RpcClientFactory.getDefaultInstance(flumeHostname, flumePort);
            }
        }
    }

    public void sendBatchDataToFlume(String data, String topic, Map<String, String> attachData) {

        logger.info("topic=" + topic);
        // Create a Flume Event object that encapsulates the data
//			//如果附加数据不为空，那么就把他解开放入到data中
        data = parseAttachData(attachData, data);
        logger.info("data=" + data);

        Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
        event.setHeaders(ImmutableMap.of("topic", topic));
        // Send the batch event
        append(event);
    }

    public void sendBatchDataToFlume(String value, String topic) {
        logger.info("topic=" + topic);
        // Create a Flume Event object that encapsulates the data
        logger.info("data=" + value);
        Event event = EventBuilder.withBody(value, Charset.forName("UTF-8"));
        event.setHeaders(ImmutableMap.of("topic", topic));
        append(event);
    }

    /**
     * 发送消息
     * @param event
     */
    private void append(Event event) {
        try {
            client.append(event);
        } catch (EventDeliveryException e) {
            e.printStackTrace();
            logger.error("flume client error, ready to get new instance!");
            synchronized (this) {
                client.close();
                client = RpcClientFactory.getDefaultInstance(flumeHostname, flumePort);
            }
        }
    }

    /**
     * 分装额外的数据
     *
     * @param attachData
     * @param data
     * @return
     */
    private String parseAttachData(Map<String, String> attachData, String data) {
        if (attachData != null) {
            JSONObject jsonObj = JSON.parseObject(data);
            jsonObj.putAll(attachData);
            data = JSON.toJSONString(jsonObj);
        }
        return data;
    }

    @PreDestroy
    public void cleanUp() {
        // Close the RPC connection
        if (client != null && client.isActive()) {
            client.close();
        }
    }

}
